package net.bw.realtime.game.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/*
 * @ Author：liuyawei
 * @ Date：2025-06-13
 */
public class MongoDBToStarRocksJob {

    public static void main(String[] args) {

        // 1.创建表环境
        TableEnvironment tabEnv = getTableEnv();

        // 2. 输入表-input：映射到kafka消息队列
        createInputTable(tabEnv);

        // 3. 输出表-output：映射到starrocks表
        createOutputTable(tabEnv) ;

        // 4. 保存数据-insert
        saveToStarRocks(tabEnv) ;

    }

    // todo 4. 保存数据-insert
    private static void saveToStarRocks(TableEnvironment tabEnv) {

        // 保存数据
        tabEnv.executeSql(
                "INSERT INTO game_sales_details (\n" +
                        "  sale_id,\n" +
                        "  game_id,\n" +
                        "  platform_id,\n" +
                        "  sale_date,\n" +
                        "  units_sold,\n" +
                        "  sale_amt,\n" +
                        "  game_name,\n" +
                        "  release_date,\n" +
                        "  developer,\n" +
                        "  publisher,\n" +
                        "  platform_name,\n" +
                        "  type\n" +
                        ")\n" +
                        "select\n" +
                        "  gsf.sale_id,\n" +
                        "  gs.game_id,\n" +
                        "  gs.platform_id,\n" +
                        "  gs.sale_date,\n" +
                        "  gs.units_sold,\n" +
                        "  gs.sale_amt,\n" +
                        "  gd.game_name,\n" +
                        "  gd.release_date,\n" +
                        "  gd.developer,\n" +
                        "  gd.publisher,\n" +
                        "  pd.platform_name,\n" +
                        "  pd.type\n" +
                        "from game_sales_fact as gsf\n" +
                        "join game_sales FOR SYSTEM_TIME AS OF gsf.process_time as gs\n" +
                        "on gsf.sale_id = gs.sale_id\n" +
                        "\n" +
                        "join game_dimension FOR SYSTEM_TIME AS OF gs.process_time as gd\n" +
                        "on gs.game_id = gd.game_id\n" +
                        "\n" +
                        "join platform_dimension FOR SYSTEM_TIME AS OF gs.process_time as pd\n" +
                        "on gs.platform_id = pd.platform_id"
        );

    }

    // todo 3. 输出表-output：映射到starrocks表
    private static void createOutputTable(TableEnvironment tabEnv) {

        // 创建输出表
        tabEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS game_sales_details (\n" +
                        "  sale_id INT,\n" +
                        "  game_id INT,\n" +
                        "  platform_id INT,\n" +
                        "  sale_date STRING,\n" +
                        "  units_sold INT,\n" +
                        "  sale_amt DOUBLE,\n" +
                        "  game_name STRING,\n" +
                        "  release_date STRING,\n" +
                        "  developer STRING,\n" +
                        "  publisher STRING,\n" +
                        "  platform_name STRING,\n" +
                        "  type STRING\n" +
                        ") WITH (\n" +
                        "  'connector' = 'starrocks',\n" +
                        "  'jdbc-url' = 'jdbc:mysql://node101:9030/starrocks_db',\n" +
                        "  'load-url' = 'node101:8030',\n" +
                        "  'database-name' = 'starrocks_db',\n" +
                        "  'table-name' = 'game_sales_details',\n" +
                        "  'username' = 'root',\n" +
                        "  'password' = '123456',\n" +
                        "  'sink.properties.format' = 'json',\n" +  // 添加格式
                        "  'sink.properties.strip_outer_array' = 'true',\n" +  // 处理JSON数组
                        "  'sink.buffer-flush.interval-ms' = '5000'\n" +  // 刷新间隔
                        ")"
        );

    }

    // todo 2. 输入表-input：映射到kafka消息队列
    private static void createInputTable(TableEnvironment tabEnv) {

        // 创建输入表
        tabEnv.executeSql(
                "CREATE TABLE game_sales_fact (\n" +
                        "  sale_id      INT,\n" +
                        "  process_time AS PROCTIME()\n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "  'topic' = 'game_sales_fact',    --消费的topic\n" +
                        "  'format' = 'json',              --数据格式\n" +
                        "  'properties.group.id' = 'game_sales_fact',  --消费者组\n" +
                        "  'properties.auto.offset.reset' = 'earliest' --如果消费者组首次使用，则从最早位点开始消费\n" +
                        ");"
        );

        // 游戏销售表
        tabEnv.executeSql(
                "CREATE TABLE game_sales\n" +
                        "(\n" +
                        "  `_id`       STRING,\n" +
                        "  sale_id     INT,\n" +
                        "  game_id     INT,\n" +
                        "  platform_id INT,\n" +
                        "  sale_date   STRING,\n" +
                        "  units_sold  INT,\n" +
                        "  sale_amt    DOUBLE,\n" +
                        "  process_time AS PROCTIME() \n" +
                        ")\n" +
                        "WITH (\n" +
                        "  'connector' = 'mongodb',\n" +
                        "  'uri' = 'mongodb://localhost:27017',\n" +
                        "  'database' = 'db',\n" +
                        "  'collection' = 'game_sales'\n" +
                        ")"
        );

        // 输出
        //tabEnv.executeSql("SELECT * FROM game_sales limit 20").print();

        // 游戏维度表
        tabEnv.executeSql(
                "CREATE TABLE game_dimension\n" +
                        "(\n" +
                        "  `_id`        STRING,\n" +
                        "  game_id      INT,\n" +
                        "  game_name    STRING,\n" +
                        "  release_date STRING,\n" +
                        "  developer    STRING,\n" +
                        "  publisher    STRING,\n" +
                        "  process_time AS PROCTIME()\n" +
                        ")\n" +
                        "WITH (\n" +
                        "  'connector' = 'mongodb',\n" +
                        "  'uri' = 'mongodb://localhost:27017',\n" +
                        "  'database' = 'db',\n" +
                        "  'collection' = 'game_dimension'\n" +
                        ")"
        );

        // 输出
        //tabEnv.executeSql("SELECT * FROM game_dimension limit 20").print();

        // 平台维度表
        tabEnv.executeSql(
                "CREATE TABLE platform_dimension\n" +
                        "(\n" +
                        "  `_id`          STRING\n" +
                        "  ,platform_id   INT\n" +
                        "  ,platform_name STRING\n" +
                        "  ,type          STRING\n" +
                        "  ,process_time AS PROCTIME()\n" +
                        ")\n" +
                        "WITH (\n" +
                        "  'connector' = 'mongodb',\n" +
                        "  'uri' = 'mongodb://localhost:27017',\n" +
                        "  'database' = 'db',\n" +
                        "  'collection' = 'platform_dimension'\n" +
                        ");"
        );

        // 输出
        //tabEnv.executeSql("SELECT * FROM platform_dimension limit 20").print();

    }

    // todo 1.创建表环境
    private static TableEnvironment getTableEnv() {

        // 1.创建表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();

        TableEnvironment tabEnv = TableEnvironment.create(settings);

        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");

        // 3返回对象
        return tabEnv;

    }

}
